package cn.wei.youvierpc.registry;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.ConcurrentHashSet;
import cn.hutool.cron.CronUtil;
import cn.hutool.cron.task.Task;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import cn.wei.youvierpc.config.RegistryConfig;
import cn.wei.youvierpc.model.ServiceMetaInfo;
import io.etcd.jetcd.*;
import io.etcd.jetcd.kv.GetResponse;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.PutOption;
import io.etcd.jetcd.watch.WatchEvent;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

public class EtcdRegistry implements Registry{

    private final Set<String> etcdRegisterNodeKeys = new HashSet<>();

    private final Set<String> watchingKeySet = new ConcurrentHashSet();

    /*public static void main(String[] args) throws ExecutionException, InterruptedException {
        Client client = Client.builder().endpoints("http://localhost:2379")
                .build();

        KV kvClient = client.getKVClient();
        ByteSequence key = ByteSequence.from("test_key".getBytes());
        ByteSequence value = ByteSequence.from("test_value".getBytes());

        kvClient.put(key, value).get();

        CompletableFuture<GetResponse> getFuture = kvClient.get(key);
        GetResponse response = getFuture.get();
        kvClient.delete(key).get();
    }*/

    private Client client;

    private KV kvClient;

    private final RegistryServiceCache serviceCache = new RegistryServiceCache();

    private static final String ETCD_ROOT_PATH = "/rpc/";

    @Override
    public void init(RegistryConfig registryConfig) {
        client = Client.builder().endpoints(registryConfig.getAddress()).connectTimeout(Duration.ofMillis(registryConfig.getTimeout()))
                .build();

        kvClient = client.getKVClient();

        heartBeat();

    }

    @Override
    public void register(ServiceMetaInfo serviceMetaInfo) throws Exception {
        Lease leaseClient = client.getLeaseClient();

        long leaseId = leaseClient.grant(30).get().getID();

        String registerKey = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        ByteSequence key = ByteSequence.from(registerKey, StandardCharsets.UTF_8);
        ByteSequence value = ByteSequence.from(JSONUtil.toJsonStr(serviceMetaInfo), StandardCharsets.UTF_8);

        PutOption putOption = PutOption.builder().withLeaseId(leaseId).build();

        kvClient.put(key, value, putOption).get();

        //本地记录
        etcdRegisterNodeKeys.add(registerKey);
    }

    @Override
    public void unRegister(ServiceMetaInfo serviceMetaInfo) {
        String registerKey = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        kvClient.delete(ByteSequence.from(registerKey,
                StandardCharsets.UTF_8));

        etcdRegisterNodeKeys.remove(registerKey);
    }

    @Override
    public List<ServiceMetaInfo> serviceDiscovery(String serviceKey) {
        //优先从本地缓存里面获取服务信息
        List<ServiceMetaInfo> serviceMetaInfos = serviceCache.readCache();
        if (serviceMetaInfos != null) {
            return serviceMetaInfos;
        }

        String searchPrefix = ETCD_ROOT_PATH + serviceKey + "/";

        GetOption getOption = GetOption.builder().isPrefix(true).build();

        List<KeyValue> keyValues = null;
        try {
            keyValues = kvClient.get(
                    ByteSequence.from(searchPrefix, StandardCharsets.UTF_8),
                    getOption
            ).get().getKvs();
        } catch (Exception e) {
            throw new RuntimeException("获取服务列表失败", e);
        }

        return keyValues.stream()
                .map(keyValue -> {
                    //监听对应服务
                    String key = keyValue.getKey().toString(StandardCharsets.UTF_8);
                    watch(key);

                    String value = keyValue.getValue().toString(StandardCharsets.UTF_8);
                    return JSONUtil.toBean(value, ServiceMetaInfo.class);
                }).collect(Collectors.toList());
    }

    @Override
    public void destroy() {
        System.out.println("当前节点下线");

        for (String etcdRegisterNodeKey : etcdRegisterNodeKeys) {
            try {
                kvClient.delete(ByteSequence.from(etcdRegisterNodeKey, StandardCharsets.UTF_8)).get();
            } catch (Exception e) {
                throw new RuntimeException(etcdRegisterNodeKey + "节点下线失败");
            }
        }

        if (kvClient != null) {
            kvClient.close();
        }

        if (client != null) {
            client.close();
        }

    }

    @Override
    public void heartBeat() {

        CronUtil.schedule("*/10 * * * * *", new Task() {
            @Override
            public void execute() {
                for (String etcdRegisterKey : etcdRegisterNodeKeys) {
                    try {
                        List<KeyValue> kvs = kvClient.get(ByteSequence.from(etcdRegisterKey, StandardCharsets.UTF_8))
                                .get()
                                .getKvs();

                        if (CollUtil.isEmpty(kvs)) {
                            continue;
                        }

                        KeyValue firstKV = kvs.get(0);
                        String value = firstKV.getValue().toString(StandardCharsets.UTF_8);
                        ServiceMetaInfo serviceMetaInfo = JSONUtil.toBean(value, ServiceMetaInfo.class);
                        register(serviceMetaInfo);
                    } catch (Exception e) {
                        throw new RuntimeException(etcdRegisterKey + "续签失败", e);
                    }
                }
            }
        });

        CronUtil.setMatchSecond(true);
        CronUtil.start();
    }


    /**
     * 消费端获取到对应的服务信息后使用此方法进行监听
     * @param serviceNodeKey
     */
    @Override
    public void watch(String serviceNodeKey) {

        Watch watchClient = client.getWatchClient();
        boolean newWatch = watchingKeySet.add(serviceNodeKey);
        if (newWatch) {
            watchClient.watch(ByteSequence.from(serviceNodeKey, StandardCharsets.UTF_8), watchResponse -> {
                for (WatchEvent event : watchResponse.getEvents()) {
                    switch (event.getEventType()) {
                        case DELETE -> serviceCache.clearCache();
                    }
                }
            });
        }
    }
}
